/**
 * MemCached Java client, connection pool for Socket IO
 * Copyright (c) 2007 Greg Whalin
 * All rights reserved.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the BSD license
 *
 * This library is distributed in the hope that it will be
 * useful, but WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.
 *
 * You should have received a copy of the BSD License along with this
 * library.
 *
 * @author greg whalin <greg@meetup.com> 
 * @author Richard 'toast' Russo <russor@msoe.edu>
 * @version 1.5.2
 */
package com.painiu.pnfs;

import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * This class is a connection pool for maintaning a pool of persistent
 * connections<br/> to memcached servers.
 * 
 * The pool must be initialized prior to use. This should typically be early on<br/>
 * in the lifecycle of the JVM instance.<br/> <br/>
 * <h3>An example of initializing using defaults:</h3>
 * 
 * <pre>
 * static {
 * 	String[] serverlist = { &quot;cache0.server.com:12345&quot;,
 * 			&quot;cache1.server.com:12345&quot; };
 * 
 * 	SockIOPool pool = SockIOPool.getInstance();
 * 	pool.setServers(serverlist);
 * 	pool.initialize();
 * }
 * </pre>
 * 
 * <h3>An example of initializing using defaults and providing weights for
 * servers:</h3>
 * 
 * <pre>
 * static {
 * 	String[] serverlist = { &quot;cache0.server.com:12345&quot;,
 * 			&quot;cache1.server.com:12345&quot; };
 * 	Integer[] weights = { new Integer(5), new Integer(2) };
 * 
 * 	SockIOPool pool = SockIOPool.getInstance();
 * 	pool.setServers(serverlist);
 * 	pool.setWeights(weights);
 * 	pool.initialize();
 * }
 * </pre>
 * 
 * <h3>An example of initializing overriding defaults:</h3>
 * 
 * <pre>
 * static {
 * 	String[] serverlist = { &quot;cache0.server.com:12345&quot;,
 * 			&quot;cache1.server.com:12345&quot; };
 * 	Integer[] weights = { new Integer(5), new Integer(2) };
 * 	int initialConnections = 10;
 * 	int minSpareConnections = 5;
 * 	int maxSpareConnections = 50;
 * 	long maxIdleTime = 1000 * 60 * 30; // 30 minutes
 * 	long maxBusyTime = 1000 * 60 * 5; // 5 minutes
 * 	long maintThreadSleep = 1000 * 5; // 5 seconds
 * 	int socketTimeOut = 1000 * 3; // 3 seconds to block on reads
 * 	int socketConnectTO = 1000 * 3; // 3 seconds to block on initial connections.  If 0, then will use blocking connect (default)
 * 	boolean failover = false; // turn off auto-failover in event of server down	
 * 	boolean nagleAlg = false; // turn off Nagle's algorithm on all sockets in pool	
 * 	boolean aliveCheck = false; // disable health check of socket on checkout
 * 
 * 	SockIOPool pool = SockIOPool.getInstance();
 * 	pool.setServers(serverlist);
 * 	pool.setWeights(weights);
 * 	pool.setInitConn(initialConnections);
 * 	pool.setMinConn(minSpareConnections);
 * 	pool.setMaxConn(maxSpareConnections);
 * 	pool.setMaxIdle(maxIdleTime);
 * 	pool.setMaxBusyTime(maxBusyTime);
 * 	pool.setMaintSleep(maintThreadSleep);
 * 	pool.setSocketTO(socketTimeOut);
 * 	pool.setNagle(nagleAlg);
 * 	pool.setHashingAlg(SockIOPool.NEW_COMPAT_HASH);
 * 	pool.setAliveCheck(true);
 * 	pool.initialize();
 * }
 * </pre>
 * 
 * The easiest manner in which to initialize the pool is to set the servers and
 * rely on defaults as in the first example.<br/> After pool is initialized, a
 * client will request a SockIO object by calling getSock with the cache key<br/>
 * The client must always close the SockIO object when finished, which will
 * return the connection back to the pool.<br/>
 * <h3>An example of retrieving a SockIO object:</h3>
 * 
 * <pre>
 * 		SockIOPool.SockIO sock = SockIOPool.getInstance().getSock( key );
 * 		try {
 * 			sock.write( &quot;version\r\n&quot; );	
 * 			sock.flush();	
 * 			System.out.println( &quot;Version: &quot; + sock.readLine() );	
 * 		}
 * 		catch (IOException ioe) { System.out.println( &quot;io exception thrown&quot; ) };	
 * 
 * 		sock.close();	
 * </pre>
 * 
 * @author greg whalin <greg@whalin.com>
 * @version 1.5
 */
public class SockIOPool {

	// logger
	private static Log log = LogFactory.getLog(SockIOPool.class.getName());

	// store instances of pools
	private static Map<String, SockIOPool> pools = new HashMap<String, SockIOPool>();

	// Constants
	public static final int NATIVE_HASH = 0; // native String.hashCode();

	public static final int OLD_COMPAT_HASH = 1; // original compatibility
													// hashing algorithm (works
													// with other clients)

	public static final int NEW_COMPAT_HASH = 2; // new CRC32 based
													// compatibility hashing
													// algorithm (works with
													// other clients)

	public static final long MAX_RETRY_DELAY = 10 * 60 * 1000; // max of 10
																// minute delay
																// for fall off

	// Pool data
	private MaintThread maintThread;

	private boolean initialized = false;

	private int maxCreate = 1; // this will be initialized by pool when the
								// pool is initialized

	private Map<String, Integer> createShift;

	// initial, min and max pool sizes
	private int poolMultiplier = 4;

	private int initConn = 3;

	private int minConn = 3;

	private int maxConn = 50;

	private long maxIdle = 1000 * 60 * 3; // max idle time for avail sockets

	private long maxBusyTime = 1000 * 60 * 5; // max idle time for avail
												// sockets

	private long maintSleep = 1000 * 5; // maintenance thread sleep time

	private int socketTO = 1000 * 10; // default timeout of socket reads

	private int socketConnectTO = 1000 * 3; // default timeout of socket
											// connections

	private boolean aliveCheck = false; // default to not check each connection
										// for being alive

	private boolean failover = true; // default to failover in event of cache
										// server dead

	private boolean failback = true; // only used if failover is also set ...
										// controls putting a dead server back
										// into rotation

	private boolean nagle = true; // enable/disable Nagle's algorithm

//	private int hashingAlg = NATIVE_HASH; // default to using the native hash
											// as it is the fastest

	// locks
	private final ReentrantLock hostDeadLock = new ReentrantLock();

	// list of all servers
	private String[] servers;

	private Integer[] weights;

	private List<String> buckets;

	// dead server map
	private Map<String, Date> hostDead;

	private Map<String, Long> hostDeadDur;

	// map to hold all available sockets
	private Map<String, Map<SockIO, Long>> availPool;

	// map to hold busy sockets
	private Map<String, Map<SockIO, Long>> busyPool;

	// empty constructor
	protected SockIOPool() {
	}

	/**
	 * Factory to create/retrieve new pools given a unique poolName.
	 * 
	 * @param poolName
	 *            unique name of the pool
	 * @return instance of SockIOPool
	 */
	public static synchronized SockIOPool getInstance(String poolName) {
		if (pools.containsKey(poolName)) {
			return pools.get(poolName);
		}

		SockIOPool pool = new SockIOPool();
		pools.put(poolName, pool);

		return pool;
	}

	/**
	 * Single argument version of factory used for back compat. Simply creates a
	 * pool named "default".
	 * 
	 * @return instance of SockIOPool
	 */
	public static SockIOPool getInstance() {
		return getInstance("default");
	}

	/**
	 * Sets the list of all cache servers.
	 * 
	 * @param servers
	 *            String array of servers [host:port]
	 */
	public void setServers(String[] servers) {
		this.servers = servers;
	}

	/**
	 * Returns the current list of all cache servers.
	 * 
	 * @return String array of servers [host:port]
	 */
	public String[] getServers() {
		return this.servers;
	}

	/**
	 * Sets the list of weights to apply to the server list.
	 * 
	 * This is an int array with each element corresponding to an element<br/>
	 * in the same position in the server String array.
	 * 
	 * @param weights
	 *            Integer array of weights
	 */
	public void setWeights(Integer[] weights) {
		this.weights = weights;
	}

	/**
	 * Returns the current list of weights.
	 * 
	 * @return int array of weights
	 */
	public Integer[] getWeights() {
		return this.weights;
	}

	/**
	 * Sets the initial number of connections per server in the available pool.
	 * 
	 * @param initConn
	 *            int number of connections
	 */
	public void setInitConn(int initConn) {
		this.initConn = initConn;
	}

	/**
	 * Returns the current setting for the initial number of connections per
	 * server in the available pool.
	 * 
	 * @return number of connections
	 */
	public int getInitConn() {
		return this.initConn;
	}

	/**
	 * Sets the minimum number of spare connections to maintain in our available
	 * pool.
	 * 
	 * @param minConn
	 *            number of connections
	 */
	public void setMinConn(int minConn) {
		this.minConn = minConn;
	}

	/**
	 * Returns the minimum number of spare connections in available pool.
	 * 
	 * @return number of connections
	 */
	public int getMinConn() {
		return this.minConn;
	}

	/**
	 * Sets the maximum number of spare connections allowed in our available
	 * pool.
	 * 
	 * @param maxConn
	 *            number of connections
	 */
	public void setMaxConn(int maxConn) {
		this.maxConn = maxConn;
	}

	/**
	 * Returns the maximum number of spare connections allowed in available
	 * pool.
	 * 
	 * @return number of connections
	 */
	public int getMaxConn() {
		return this.maxConn;
	}

	/**
	 * Sets the max idle time for threads in the available pool.
	 * 
	 * @param maxIdle
	 *            idle time in ms
	 */
	public void setMaxIdle(long maxIdle) {
		this.maxIdle = maxIdle;
	}

	/**
	 * Returns the current max idle setting.
	 * 
	 * @return max idle setting in ms
	 */
	public long getMaxIdle() {
		return this.maxIdle;
	}

	/**
	 * Sets the max busy time for threads in the busy pool.
	 * 
	 * @param maxBusyTime
	 *            idle time in ms
	 */
	public void setMaxBusyTime(long maxBusyTime) {
		this.maxBusyTime = maxBusyTime;
	}

	/**
	 * Returns the current max busy setting.
	 * 
	 * @return max busy setting in ms
	 */
	public long getMaxBusy() {
		return this.maxBusyTime;
	}

	/**
	 * Set the sleep time between runs of the pool maintenance thread. If set to
	 * 0, then the maint thread will not be started.
	 * 
	 * @param maintSleep
	 *            sleep time in ms
	 */
	public void setMaintSleep(long maintSleep) {
		this.maintSleep = maintSleep;
	}

	/**
	 * Returns the current maint thread sleep time.
	 * 
	 * @return sleep time in ms
	 */
	public long getMaintSleep() {
		return this.maintSleep;
	}

	/**
	 * Sets the socket timeout for reads.
	 * 
	 * @param socketTO
	 *            timeout in ms
	 */
	public void setSocketTO(int socketTO) {
		this.socketTO = socketTO;
	}

	/**
	 * Returns the socket timeout for reads.
	 * 
	 * @return timeout in ms
	 */
	public int getSocketTO() {
		return this.socketTO;
	}

	/**
	 * Sets the socket timeout for connect.
	 * 
	 * @param socketConnectTO
	 *            timeout in ms
	 */
	public void setSocketConnectTO(int socketConnectTO) {
		this.socketConnectTO = socketConnectTO;
	}

	/**
	 * Returns the socket timeout for connect.
	 * 
	 * @return timeout in ms
	 */
	public int getSocketConnectTO() {
		return this.socketConnectTO;
	}

	/**
	 * Sets the failover flag for the pool.
	 * 
	 * If this flag is set to true, and a socket fails to connect,<br/> the
	 * pool will attempt to return a socket from another server<br/> if one
	 * exists. If set to false, then getting a socket<br/> will return null if
	 * it fails to connect to the requested server.
	 * 
	 * @param failover
	 *            true/false
	 */
	public void setFailover(boolean failover) {
		this.failover = failover;
	}

	/**
	 * Returns current state of failover flag.
	 * 
	 * @return true/false
	 */
	public boolean getFailover() {
		return this.failover;
	}

	/**
	 * Sets the failback flag for the pool.
	 * 
	 * If this is true and we have marked a host as dead, will try to bring it
	 * back. If it is false, we will never try to resurrect a dead host.
	 * 
	 * @param failback
	 *            true/false
	 */
	public void setFailback(boolean failback) {
		this.failback = failback;
	}

	/**
	 * Returns current state of failover flag.
	 * 
	 * @return true/false
	 */
	public boolean getFailback() {
		return this.failback;
	}

	/**
	 * Sets the aliveCheck flag for the pool.
	 * 
	 * When true, this will attempt to talk to the server on every connection
	 * checkout to make sure the connection is still valid. This adds extra
	 * network chatter and thus is defaulted off. May be useful if you want to
	 * ensure you do not have any problems talking to the server on a dead
	 * connection.
	 * 
	 * @param aliveCheck
	 *            true/false
	 */
	public void setAliveCheck(boolean aliveCheck) {
		this.aliveCheck = aliveCheck;
	}

	/**
	 * Returns the current status of the aliveCheck flag.
	 * 
	 * @return true / false
	 */
	public boolean getAliveCheck() {
		return this.aliveCheck;
	}

	/**
	 * Sets the Nagle alg flag for the pool.
	 * 
	 * If false, will turn off Nagle's algorithm on all sockets created.
	 * 
	 * @param nagle
	 *            true/false
	 */
	public void setNagle(boolean nagle) {
		this.nagle = nagle;
	}

	/**
	 * Returns current status of nagle flag
	 * 
	 * @return true/false
	 */
	public boolean getNagle() {
		return this.nagle;
	}

	/**
	 * Initializes the pool.
	 */
	public void initialize() {

		synchronized (this) {

			// check to see if already initialized
			if (initialized && (buckets != null) && (availPool != null)
					&& (busyPool != null)) {
				log.error("++++ trying to initialize an already initialized pool");
				return;
			}

			// initialize empty maps
			buckets = new ArrayList<String>();
			availPool = new HashMap<String, Map<SockIO, Long>>(servers.length
					* initConn);
			busyPool = new HashMap<String, Map<SockIO, Long>>(servers.length
					* initConn);
			hostDeadDur = new HashMap<String, Long>();
			hostDead = new HashMap<String, Date>();
			createShift = new HashMap<String, Integer>();
			maxCreate = (poolMultiplier > minConn) ? minConn : minConn
					/ poolMultiplier; // only create up to maxCreate
										// connections at once

			log.debug("++++ initializing pool with following settings:");
			log.debug("++++ initial size: " + initConn);
			log.debug("++++ min spare   : " + minConn);
			log.debug("++++ max spare   : " + maxConn);

			// if servers is not set, or it empty, then
			// throw a runtime exception
			if (servers == null || servers.length <= 0) {
				log.error("++++ trying to initialize with no servers");
				throw new IllegalStateException(
						"++++ trying to initialize with no servers");
			}

			for (int i = 0; i < servers.length; i++) {

				// add to bucket
				// with weights if we have them
				if (weights != null && weights.length > i) {
					for (int k = 0; k < weights[i].intValue(); k++) {
						buckets.add(servers[i]);
						log.debug("++++ added " + servers[i]
								+ " to server bucket");
					}
				} else {
					buckets.add(servers[i]);
					log.debug("++++ added " + servers[i] + " to server bucket");
				}

				// create initial connections
				log.debug("+++ creating initial connections (" + initConn
						+ ") for host: " + servers[i]);

				for (int j = 0; j < initConn; j++) {
					SockIO socket = createSocket(servers[i]);
					if (socket == null) {
						log.error("++++ failed to create connection to: "
								+ servers[i] + " -- only " + j + " created.");
						break;
					}

					addSocketToPool(availPool, servers[i], socket);
					log.debug("++++ created and added socket: "
							+ socket.toString() + " for host " + servers[i]);
				}
			}

			// mark pool as initialized
			this.initialized = true;

			// start maint thread
			if (this.maintSleep > 0)
				this.startMaintThread();
		}
	}

	/**
	 * Returns state of pool.
	 * 
	 * @return <CODE>true</CODE> if initialized.
	 */
	public boolean isInitialized() {
		return initialized;
	}

	/**
	 * Creates a new SockIO obj for the given server.
	 * 
	 * If server fails to connect, then return null and do not try<br/> again
	 * until a duration has passed. This duration will grow<br/> by doubling
	 * after each failed attempt to connect.
	 * 
	 * @param host
	 *            host:port to connect to
	 * @return SockIO obj or null if failed to create
	 */
	protected SockIO createSocket(String host) {

		SockIO socket = null;

		// if host is dead, then we don't need to try again
		// until the dead status has expired
		// we do not try to put back in if failback is off

		hostDeadLock.lock();
		try {
			if (failback && hostDead.containsKey(host)
					&& hostDeadDur.containsKey(host)) {

				Date store = hostDead.get(host);
				long expire = hostDeadDur.get(host).longValue();

				if ((store.getTime() + expire) > System.currentTimeMillis()) {
					return null;
				}
			}
		} finally {
			hostDeadLock.unlock();
		}

		try {
			socket = new SockIO(this, host, this.socketTO, this.socketConnectTO, this.nagle);

			if (!socket.isConnected()) {
				log.error("++++ failed to get SockIO obj for: " + host + " -- new socket is not connected");
				try {
					socket.trueClose();
				} catch (Exception ex) {
					log.error("++++ failed to close SockIO obj for server: " + host);
					log.error(ex.getMessage(), ex);
					socket = null;
				}
			}
		} catch (Exception ex) {
			log.error("++++ failed to get SockIO obj for: " + host);
			log.error(ex.getMessage(), ex);
			socket = null;
		}

		// if we failed to get socket, then mark
		// host dead for a duration which falls off
		hostDeadLock.lock();
		try {
			if (socket == null) {
				Date now = new Date();
				hostDead.put(host, now);

				long expire = (hostDeadDur.containsKey(host)) ? (hostDeadDur.get(host).longValue() * 2) : 1000;

				if (expire > MAX_RETRY_DELAY) {
					expire = MAX_RETRY_DELAY;
				}

				hostDeadDur.put(host, new Long(expire));
				log.debug("++++ ignoring dead host: " + host + " for " + expire + " ms");

				// also clear all entries for this host from availPool
				clearHostFromPool(availPool, host);
			} else {
				log.debug("++++ created socket (" + socket.toString() + ") for host: " + host);
				if (hostDead.containsKey(host) || hostDeadDur.containsKey(host)) {
					hostDead.remove(host);
					hostDeadDur.remove(host);
				}
			}
		} finally {
			hostDeadLock.unlock();
		}

		return socket;
	}

	/**
	 * Returns appropriate SockIO object given string cache key and optional
	 * hashcode.
	 * 
	 * Trys to get SockIO from pool. Fails over to additional pools in event of
	 * server failure.
	 * 
	 * @return SockIO obj connected to server
	 */
	public SockIO getSock() {
		if (!this.initialized) {
			log.error("attempting to get SockIO from uninitialized pool!");
			return null;
		}

		// if no servers return null
		if (buckets.size() == 0)
			return null;

		// if only one server, return it
		if (buckets.size() == 1) {
			SockIO sock = getConnection(buckets.get(0));
			if (sock != null && sock.isConnected()) {
				if (aliveCheck) {
					if (!sock.isAlive()) {
						sock.close();
						try {
							sock.trueClose();
						} catch (IOException ioe) {
							log.error("failed to close dead socket");
						}
						sock = null;
					}
				}
			} else {
				sock = null;
			}

			return sock;
		}

		// from here on, we are working w/ multiple servers
		// keep trying different servers until we find one
		int bucketSize = buckets.size();

		// get initial bucket
		int bucket = (int) (Math.random() * bucketSize);

		int tries = 0;
		while (tries++ < bucketSize) {

			// try to get socket from bucket
			SockIO sock = getConnection(buckets.get(bucket));

			log.debug("host choose: " + buckets.get(bucket));

			if (sock != null && sock.isConnected()) {
				if (aliveCheck) {
					if (sock.isAlive()) {
						return sock;
					}
					sock.close();
					try {
						sock.trueClose();
					} catch (IOException ioe) {
						log.error("failed to close dead socket");
					}
					sock = null;
				} else {
					return sock;
				}
			} else {
				sock = null;
			}

			// if we do not want to failover, then bail here
			if (!failover) {
				return null;
			}
			
			if (++bucket == bucketSize) {
				bucket = 0;
			}
		}

		return null;
	}

	/**
	 * Returns a SockIO object from the pool for the passed in host.
	 * 
	 * Meant to be called from a more intelligent method<br/> which handles
	 * choosing appropriate server<br/> and failover.
	 * 
	 * @param host
	 *            host from which to retrieve object
	 * @return SockIO object or null if fail to retrieve one
	 */
	public SockIO getConnection(String host) {

		if (!this.initialized) {
			log.error("attempting to get SockIO from uninitialized pool!");
			return null;
		}

		if (host == null) {
			return null;
		}

		synchronized (this) {

			// if we have items in the pool
			// then we can return it
			if (availPool != null && !availPool.isEmpty()) {

				// take first connected socket
				Map<SockIO, Long> aSockets = availPool.get(host);

				if (aSockets != null && !aSockets.isEmpty()) {

					for (Iterator<SockIO> i = aSockets.keySet().iterator(); i.hasNext();) {
						SockIO socket = i.next();

						if (socket.isConnected()) {
							log.debug("++++ moving socket for host (" + host + ") to busy pool ... socket: " + socket);

							// remove from avail pool
							i.remove();

							// add to busy pool
							addSocketToPool(busyPool, host, socket);

							// return socket
							return socket;
						}
						
						// not connected, so we need to remove it
						try {
							socket.trueClose();
						} catch (Exception ex) {
							log.debug("++++ error trying to true close the socket");
						}

						socket = null;

						// remove from avail pool
						i.remove();
					}
				}
			}

			// if here, then we found no sockets in the pool
			// try to create on a sliding scale up to maxCreate
			Integer cShift = createShift.get(host);
			int shift = (cShift != null) ? cShift.intValue() : 0;

			int create = 1 << shift;
			if (create >= maxCreate) {
				create = maxCreate;
			} else {
				shift++;
			}

			// store the shift value for this host
			createShift.put(host, new Integer(shift));

			log.debug("++++ creating " + create + " new SockIO objects");

			for (int i = create; i > 0; i--) {
				SockIO socket = createSocket(host);
				if (socket == null) {
					break;
				}

				if (i == 1) {
					// last iteration, add to busy pool and return sockio
					addSocketToPool(busyPool, host, socket);
					return socket;
				}
				// add to avail pool
				addSocketToPool(availPool, host, socket);
			}
		}

		// should never get here
		return null;
	}

	/**
	 * Adds a socket to a given pool for the given host. THIS METHOD IS NOT
	 * THREADSAFE, SO BE CAREFUL WHEN USING!
	 * 
	 * Internal utility method.
	 * 
	 * @param pool
	 *            pool to add to
	 * @param host
	 *            host this socket is connected to
	 * @param socket
	 *            socket to add
	 */
	protected void addSocketToPool(Map<String, Map<SockIO, Long>> pool,
			String host, SockIO socket) {

		if (pool.containsKey(host)) {
			Map<SockIO, Long> sockets = pool.get(host);

			if (sockets != null) {
				sockets.put(socket, new Long(System.currentTimeMillis()));
				return;
			}
		}

		Map<SockIO, Long> sockets = new IdentityHashMap<SockIO, Long>();

		sockets.put(socket, new Long(System.currentTimeMillis()));
		pool.put(host, sockets);
	}

	/**
	 * Removes a socket from specified pool for host. THIS METHOD IS NOT
	 * THREADSAFE, SO BE CAREFUL WHEN USING!
	 * 
	 * Internal utility method.
	 * 
	 * @param pool
	 *            pool to remove from
	 * @param host
	 *            host pool
	 * @param socket
	 *            socket to remove
	 */
	protected void removeSocketFromPool(Map<String, Map<SockIO, Long>> pool,
			String host, SockIO socket) {
		if (pool.containsKey(host)) {
			Map<SockIO, Long> sockets = pool.get(host);
			if (sockets != null)
				sockets.remove(socket);
		}
	}

	/**
	 * Closes and removes all sockets from specified pool for host. THIS METHOD
	 * IS NOT THREADSAFE, SO BE CAREFUL WHEN USING!
	 * 
	 * Internal utility method.
	 * 
	 * @param pool
	 *            pool to clear
	 * @param host
	 *            host to clear
	 */
	protected void clearHostFromPool(Map<String, Map<SockIO, Long>> pool, String host) {

		if (pool.containsKey(host)) {
			Map<SockIO, Long> sockets = pool.get(host);

			if (sockets != null && sockets.size() > 0) {
				for (Iterator<SockIO> i = sockets.keySet().iterator(); i.hasNext();) {
					SockIO socket = i.next();
					try {
						socket.trueClose();
					} catch (IOException ioe) {
						log.error("++++ failed to close socket: " + ioe.getMessage());
					}

					i.remove();
					socket = null;
				}
			}
		}
	}

	/**
	 * Checks a SockIO object in with the pool.
	 * 
	 * This will remove SocketIO from busy pool, and optionally<br/> add to
	 * avail pool.
	 * 
	 * @param socket
	 *            socket to return
	 * @param addToAvail
	 *            add to avail pool if true
	 */
	public void checkIn(SockIO socket, boolean addToAvail) {

		String host = socket.getHost();
		log.debug("++++ calling check-in on socket: " + socket.toString() + " for host: " + host);

		synchronized (this) {

			// remove from the busy pool
			log.debug("++++ removing socket (" + socket.toString() + ") from busy pool for host: " + host);
			removeSocketFromPool(busyPool, host, socket);

			// add to avail pool
			if (addToAvail && socket.isConnected()) {
				log.debug("++++ returning socket (" + socket.toString() + " to avail pool for host: " + host);
				addSocketToPool(availPool, host, socket);
			}
		}
	}

	/**
	 * Returns a socket to the avail pool.
	 * 
	 * This is called from SockIO.close(). Calling this method<br/> directly
	 * without closing the SockIO object first<br/> will cause an IOException
	 * to be thrown.
	 * 
	 * @param socket
	 *            socket to return
	 */
	public void checkIn(SockIO socket) {
		checkIn(socket, true);
	}

	/**
	 * Closes all sockets in the passed in pool.
	 * 
	 * Internal utility method.
	 * 
	 * @param pool
	 *            pool to close
	 */
	protected void closePool(Map<String, Map<SockIO, Long>> pool) {
		for (Iterator<String> i = pool.keySet().iterator(); i.hasNext();) {
			String host = i.next();
			Map<SockIO, Long> sockets = pool.get(host);

			for (Iterator<SockIO> j = sockets.keySet().iterator(); j.hasNext();) {
				SockIO socket = j.next();

				try {
					socket.trueClose();
				} catch (IOException ioe) {
					log.error("++++ failed to trueClose socket: " + socket.toString() + " for host: " + host);
				}

				j.remove();
				socket = null;
			}
		}
	}

	/**
	 * Shuts down the pool.
	 * 
	 * Cleanly closes all sockets.<br/> Stops the maint thread.<br/> Nulls out
	 * all internal maps<br/>
	 */
	public void shutDown() {
		synchronized (this) {
			log.debug("++++ SockIOPool shutting down...");

			if (maintThread != null && maintThread.isRunning()) {
				// stop the main thread
				stopMaintThread();

				// wait for the thread to finish
				while (maintThread.isRunning()) {
					log.debug("++++ waiting for main thread to finish run +++");
					try {
						Thread.sleep(500);
					} catch (Exception ex) {
					}

				}
			}

			log.debug("++++ closing all internal pools.");
			closePool(availPool);
			closePool(busyPool);
			availPool = null;
			busyPool = null;
			buckets = null;
			hostDeadDur = null;
			hostDead = null;
			initialized = false;
			log.debug("++++ SockIOPool finished shutting down.");
		}
	}

	/**
	 * Starts the maintenance thread.
	 * 
	 * This thread will manage the size of the active pool<br/> as well as move
	 * any closed, but not checked in sockets<br/> back to the available pool.
	 */
	protected void startMaintThread() {

		if (maintThread != null) {

			if (maintThread.isRunning()) {
				log.error("main thread already running");
			} else {
				maintThread.start();
			}
		} else {
			maintThread = new MaintThread(this);
			maintThread.setInterval(this.maintSleep);
			maintThread.start();
		}
	}

	/**
	 * Stops the maintenance thread.
	 */
	protected void stopMaintThread() {
		if (maintThread != null && maintThread.isRunning())
			maintThread.stopThread();
	}

	/**
	 * Runs self maintenance on all internal pools.
	 * 
	 * This is typically called by the maintenance thread to manage pool size.
	 */
	protected void selfMaint() {
		log.debug("++++ Starting self maintenance....");

		synchronized (this) {

			// go through avail sockets and create/destroy sockets
			// as needed to maintain pool settings
			for (Iterator<String> i = availPool.keySet().iterator(); i
					.hasNext();) {
				String host = i.next();
				Map<SockIO, Long> sockets = availPool.get(host);
				log.debug("++++ Size of avail pool for host (" + host + ") = " + sockets.size());

				// if pool is too small (n < minSpare)
				if (sockets.size() < minConn) {
					// need to create new sockets
					int need = minConn - sockets.size();
					log.debug("++++ Need to create " + need + " new sockets for pool for host: " + host);

					for (int j = 0; j < need; j++) {
						SockIO socket = createSocket(host);

						if (socket == null)
							break;

						addSocketToPool(availPool, host, socket);
					}
				} else if (sockets.size() > maxConn) {
					// need to close down some sockets
					int diff = sockets.size() - maxConn;
					int needToClose = (diff <= poolMultiplier) ? diff : (diff) / poolMultiplier;

					log.debug("++++ need to remove " + needToClose + " spare sockets for pool for host: " + host);
					for (Iterator<SockIO> j = sockets.keySet().iterator(); j.hasNext();) {

						if (needToClose <= 0)
							break;

						// remove stale entries
						SockIO socket = j.next();
						long expire = sockets.get(socket).longValue();

						// if past idle time
						// then close socket
						// and remove from pool
						if ((expire + maxIdle) < System.currentTimeMillis()) {
							log.debug("+++ removing stale entry from pool as it is past its idle timeout and pool is over max spare");
							try {
								socket.trueClose();
							} catch (IOException ioe) {
								log.error("failed to close socket");
								log.error(ioe.getMessage(), ioe);
							}

							j.remove();
							socket = null;
							needToClose--;
						}
					}
				}

				// reset the shift value for creating new SockIO objects
				createShift.put(host, new Integer(0));
			}

			// go through busy sockets and destroy sockets
			// as needed to maintain pool settings
			for (Iterator<String> i = busyPool.keySet().iterator(); i.hasNext();) {

				String host = i.next();
				Map<SockIO, Long> sockets = busyPool.get(host);

				log.debug("++++ Size of busy pool for host (" + host + ")  = " + sockets.size());

				// loop through all connections and check to see if we have any
				// hung connections
				for (Iterator<SockIO> j = sockets.keySet().iterator(); j.hasNext();) {
					// remove stale entries
					SockIO socket = j.next();
					long hungTime = sockets.get(socket).longValue();

					// if past max busy time
					// then close socket
					// and remove from pool
					if ((hungTime + maxBusyTime) < System.currentTimeMillis()) {
						log.error("+++ removing potentially hung connection from busy pool ... socket in pool for "
										+ (System.currentTimeMillis() - hungTime)
										+ "ms");
						try {
							socket.trueClose();
						} catch (IOException ioe) {
							log.error("failed to close socket");
							log.error(ioe.getMessage(), ioe);
						}

						j.remove();
						socket = null;
					}
				}
			}
		}

		log.debug("+++ ending self maintenance.");
	}

	/**
	 * Class which extends thread and handles maintenance of the pool.
	 * 
	 * @author greg whalin <greg@meetup.com>
	 * @version 1.5
	 */
	protected static class MaintThread extends Thread {

		private SockIOPool pool;

		private long interval = 1000 * 3; // every 3 seconds

		private boolean stopThread = false;

		private boolean running;

		protected MaintThread(SockIOPool pool) {
			this.pool = pool;
			this.setDaemon(true);
		}

		public void setInterval(long interval) {
			this.interval = interval;
		}

		public boolean isRunning() {
			return this.running;
		}

		/**
		 * sets stop variable and interupts any wait
		 */
		public void stopThread() {
			this.stopThread = true;
			this.interrupt();
		}

		/**
		 * Start the thread.
		 */
		public void run() {
			this.running = true;

			while (!this.stopThread) {
				try {
					Thread.sleep(interval);

					// if pool is initialized, then
					// run the maintenance method on itself
					if (pool.isInitialized())
						pool.selfMaint();

				} catch (Exception e) {
					break;
				}
			}

			this.running = false;
		}
	}

	/**
	 * MemCached Java client, utility class for Socket IO.
	 * 
	 * This class is a wrapper around a Socket and its streams.
	 * 
	 * @author greg whalin <greg@meetup.com>
	 * @author Richard 'toast' Russo <russor@msoe.edu>
	 * @version 1.5
	 */
	public static class SockIO {

		// logger
		private static Log log = LogFactory.getLog(SockIO.class.getName());

		// pool
		private SockIOPool pool;

		// data
		private String host;

		private Socket sock;

		private DataInputStream in;

		private BufferedOutputStream out;

		/**
		 * creates a new SockIO object wrapping a socket connection to
		 * host:port, and its input and output streams
		 * 
		 * @param pool
		 *            Pool this object is tied to
		 * @param host
		 *            host to connect to
		 * @param port
		 *            port to connect to
		 * @param timeout
		 *            int ms to block on data for read
		 * @param connectTimeout
		 *            timeout (in ms) for initial connection
		 * @param noDelay
		 *            TCP NODELAY option?
		 * @throws IOException
		 *             if an io error occurrs when creating socket
		 * @throws UnknownHostException
		 *             if hostname is invalid
		 */
		public SockIO(SockIOPool pool, String host, int port, int timeout,
				int connectTimeout, boolean noDelay) throws IOException,
				UnknownHostException {

			this.pool = pool;

			sock = getSocket(host, port, connectTimeout);

			if (timeout >= 0)
				sock.setSoTimeout(timeout);

			// testing only
			sock.setTcpNoDelay(noDelay);

			// wrap streams
			in = new DataInputStream(sock.getInputStream());
			out = new BufferedOutputStream(sock.getOutputStream());

			this.host = host + ":" + port;
		}

		/**
		 * creates a new SockIO object wrapping a socket connection to
		 * host:port, and its input and output streams
		 * 
		 * @param host
		 *            hostname:port
		 * @param timeout
		 *            read timeout value for connected socket
		 * @param connectTimeout
		 *            timeout for initial connections
		 * @param noDelay
		 *            TCP NODELAY option?
		 * @throws IOException
		 *             if an io error occurrs when creating socket
		 * @throws UnknownHostException
		 *             if hostname is invalid
		 */
		public SockIO(SockIOPool pool, String host, int timeout,
				int connectTimeout, boolean noDelay) throws IOException,
				UnknownHostException {

			this.pool = pool;

			String[] ip = host.split(":");

			// get socket: default is to use non-blocking connect
			sock = getSocket(ip[0], Integer.parseInt(ip[1]), connectTimeout);

			if (timeout >= 0)
				sock.setSoTimeout(timeout);

			// testing only
			sock.setTcpNoDelay(noDelay);

			// wrap streams
			in = new DataInputStream(sock.getInputStream());
			out = new BufferedOutputStream(sock.getOutputStream());
			this.host = host;
		}

		/**
		 * Method which spawns thread to get a connection and then enforces a
		 * timeout on the initial connection.
		 * 
		 * This should be backed by a thread pool. Any volunteers?
		 * 
		 * @param host
		 *            host to establish connection to
		 * @param port
		 *            port on that host
		 * @param timeout
		 *            connection timeout in ms
		 * @return connected socket
		 * @throws IOException
		 *             if errors connecting or if connection times out
		 */
		protected static Socket getSocket(String host, int port, int timeout)
				throws IOException {
			Socket sock = new Socket();
			sock.connect(new InetSocketAddress(host, port), timeout);
			return sock;
		}

		/**
		 * returns the host this socket is connected to
		 * 
		 * @return String representation of host (hostname:port)
		 */
		String getHost() {
			return this.host;
		}

		/**
		 * closes socket and all streams connected to it
		 * 
		 * @throws IOException
		 *             if fails to close streams or socket
		 */
		void trueClose() throws IOException {
			log.debug("++++ Closing socket for real: " + toString());

			boolean err = false;
			StringBuilder errMsg = new StringBuilder();

			if (in == null || out == null || sock == null) {
				err = true;
				errMsg.append("++++ socket or its streams already null in trueClose call");
			}

			if (in != null) {
				try {
					in.close();
				} catch (IOException ioe) {
					log.error("++++ error closing input stream for socket: "
							+ toString() + " for host: " + getHost());
					log.error(ioe.getMessage(), ioe);
					errMsg.append("++++ error closing input stream for socket: "
									+ toString()
									+ " for host: "
									+ getHost()
									+ "\n");
					errMsg.append(ioe.getMessage());
					err = true;
				}
			}

			if (out != null) {
				try {
					out.close();
				} catch (IOException ioe) {
					log.error("++++ error closing output stream for socket: "
							+ toString() + " for host: " + getHost());
					log.error(ioe.getMessage(), ioe);
					errMsg.append("++++ error closing output stream for socket: "
									+ toString()
									+ " for host: "
									+ getHost()
									+ "\n");
					errMsg.append(ioe.getMessage());
					err = true;
				}
			}

			if (sock != null) {
				try {
					sock.close();
				} catch (IOException ioe) {
					log.error("++++ error closing socket: " + toString()
							+ " for host: " + getHost());
					log.error(ioe.getMessage(), ioe);
					errMsg.append("++++ error closing socket: " + toString()
							+ " for host: " + getHost() + "\n");
					errMsg.append(ioe.getMessage());
					err = true;
				}
			}

			// check in to pool
			if (sock != null)
				pool.checkIn(this, false);

			in = null;
			out = null;
			sock = null;

			if (err)
				throw new IOException(errMsg.toString());
		}

		/**
		 * sets closed flag and checks in to connection pool but does not close
		 * connections
		 */
		void close() {
			// check in to pool
			log.debug("++++ marking socket (" + this.toString()
					+ ") as closed and available to return to avail pool");
			pool.checkIn(this);
		}

		/**
		 * checks if the connection is open
		 * 
		 * @return true if connected
		 */
		boolean isConnected() {
			return (sock != null && sock.isConnected());
		}

		/*
		 * checks to see that the connection is still working
		 * 
		 * @return true if still alive
		 */
		boolean isAlive() {

			if (!isConnected())
				return false;

			// try to talk to the server w/ a dumb query to ask its version
			try {
				this.write("version\r\n".getBytes());
				this.flush();
				this.readLine();
			} catch (IOException ex) {
				return false;
			}

			return true;
		}

		/**
		 * reads a line intentionally not using the deprecated readLine method
		 * from DataInputStream
		 * 
		 * @return String that was read in
		 * @throws IOException
		 *             if io problems during read
		 */
		String readLine() throws IOException {
			if (sock == null || !sock.isConnected()) {
				log.error("++++ attempting to read from closed socket");
				throw new IOException(
						"++++ attempting to read from closed socket");
			}

			byte[] b = new byte[1];
			ByteArrayOutputStream bos = new ByteArrayOutputStream();
			boolean eol = false;

			while (in.read(b, 0, 1) != -1) {

				if (b[0] == 13) {
					eol = true;
				} else {
					if (eol) {
						if (b[0] == 10)
							break;

						eol = false;
					}
				}

				// cast byte into char array
				bos.write(b, 0, 1);
			}

			if (bos == null || bos.size() <= 0) {
				throw new IOException(
						"++++ Stream appears to be dead, so closing it down");
			}

			// else return the string
			return bos.toString().trim();
		}

		/**
		 * reads up to end of line and returns nothing
		 * 
		 * @throws IOException
		 *             if io problems during read
		 */
		void clearEOL() throws IOException {
			if (sock == null || !sock.isConnected()) {
				log.error("++++ attempting to read from closed socket");
				throw new IOException(
						"++++ attempting to read from closed socket");
			}

			byte[] b = new byte[1];
			boolean eol = false;
			while (in.read(b, 0, 1) != -1) {

				// only stop when we see
				// \r (13) followed by \n (10)
				if (b[0] == 13) {
					eol = true;
					continue;
				}

				if (eol) {
					if (b[0] == 10)
						break;

					eol = false;
				}
			}
		}

		/**
		 * reads length bytes into the passed in byte array from dtream
		 * 
		 * @param b
		 *            byte array
		 * @throws IOException
		 *             if io problems during read
		 */
		void read(byte[] b) throws IOException {
			if (sock == null || !sock.isConnected()) {
				log.error("++++ attempting to read from closed socket");
				throw new IOException(
						"++++ attempting to read from closed socket");
			}

			int count = 0;
			while (count < b.length) {
				int cnt = in.read(b, count, (b.length - count));
				count += cnt;
			}
		}

		/**
		 * flushes output stream
		 * 
		 * @throws IOException
		 *             if io problems during read
		 */
		void flush() throws IOException {
			if (sock == null || !sock.isConnected()) {
				log.error("++++ attempting to write to closed socket");
				throw new IOException(
						"++++ attempting to write to closed socket");
			}
			out.flush();
		}

		/**
		 * writes a byte array to the output stream
		 * 
		 * @param b
		 *            byte array to write
		 * @throws IOException
		 *             if an io error happens
		 */
		void write(byte[] b) throws IOException {
			if (sock == null || !sock.isConnected()) {
				log.error("++++ attempting to write to closed socket");
				throw new IOException(
						"++++ attempting to write to closed socket");
			}
			out.write(b);
		}

		/**
		 * use the sockets hashcode for this object so we can key off of SockIOs
		 * 
		 * @return int hashcode
		 */
		public int hashCode() {
			return (sock == null) ? 0 : sock.hashCode();
		}

		/**
		 * returns the string representation of this socket
		 * 
		 * @return string
		 */
		public String toString() {
			return (sock == null) ? "" : sock.toString();
		}
	}
}
